package com.zhiyuancode.mqack.worker;

import com.zhiyuancode.mqack.message.base.MqResendErrorCode;
import com.zhiyuancode.mqack.message.base.MqResendRuntimeException;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ZSetOperations;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * redis操作容器
 *
 * @author 杨志远
 * email faryangsh@163.com
 * Date    2018-05-03
 */
@Getter
@Slf4j
public class RedisOperationsContext {
    private ZSetOperations<String, String> zSetOperations;

    private HashOperations<String, String, String> hashOperations;

    /**
     * 集群中当前可用服务器IP集合对应的Redis Key(按启动先后排序)
     */
    private String serverIpsKey;

    public RedisOperationsContext(ZSetOperations<String, String> zSetOperations, HashOperations<String, String, String> hashOperations, String serverIpsKey) {
        this.hashOperations = hashOperations;
        this.zSetOperations = zSetOperations;
        this.serverIpsKey = serverIpsKey;
        register();
    }

    // 本服务器IP地址
    private static String SERVER_IP;
    private static String SERVER_CHECK_LOCK_KEY = "RESEND_SERVER_HEALTH_LOCK";

    static {
        try {
            SERVER_IP = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
            SERVER_IP = "127.0.0.1";
        }
        log.info("SERVER_IP: {}", SERVER_IP);
    }


    /**
     * 启动后, 在Redis中添加本服务器的启动时间
     */
    public void register() {
        try {
            zSetOperations.add(serverIpsKey, SERVER_IP, System.currentTimeMillis());
            log.info("[{}] registered in Redis", SERVER_IP);
        } catch (Exception e) {
            log.warn("[{}] Redis Failed: {}", SERVER_IP, e.getMessage());
        }
    }

    /**
     * 启动后, 在Redis中添加本服务器的启动时间
     */
    public void refreshRegisterTime() {
        try {
            zSetOperations.add(serverIpsKey, SERVER_IP, System.currentTimeMillis());
            log.info("[{}] refresh register time in Redis", SERVER_IP);
        } catch (Exception e) {
            log.warn("[{}] Redis Failed: {}", SERVER_IP, e.getMessage());
        }
    }

    /**
     * 在Redis中删除废弃的服务器启动时间
     */
    public void unregister(String serverIp) {
        try {
            zSetOperations.remove(serverIpsKey, serverIp);
            log.info("[{}] removed from Redis", serverIp);
        } catch (Exception e) {
            log.warn("[{}] Redis Failed: {}", SERVER_IP, e.getMessage());
        }
    }
//    org.springframework.data.redis.RedisConnectionFailureException: Cannot get Jedis connection; nested exception is redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool

    /**
     * 检测集群注册健康状态
     *
     * @return true - 是最晚注册的机器
     */
    public boolean checkServerHealth() {
        try {
            return doCheckServerHealth();
        } catch (RedisConnectionFailureException e) {
            throw new MqResendRuntimeException(MqResendErrorCode.REDIS_CONNECTION_EXCEPTION, e);
        } catch (Throwable e) {
            throw new MqResendRuntimeException(MqResendErrorCode.UNKNOW_EXCEPTION, e);
        }
    }

    private boolean doCheckServerHealth() {
        // 容忍1min
        long nowMills = System.currentTimeMillis();
        long correctMills = nowMills - 60 * 1000;
        Set<String> serverIps = zSetOperations.rangeByScore(serverIpsKey, correctMills, nowMills);

        if (serverIps.size() == 1 && serverIps.contains(SERVER_IP)) {
            // 本机是集群中最后注册的服务器

            refreshRegisterTime();
            return true;
        } else if (serverIps.size() == 0) {
            // 没有机器注册/最后注册的那台机器下线，注册本机

            register();
            return false;
        } else if (serverIps.size() > 1) {
            // 集群中机器刚刚注册上去,需要推举出最晚注册的那台机器

            Set<ZSetOperations.TypedTuple<String>> typedTuples = zSetOperations.rangeWithScores(serverIpsKey, 0, -1);
            Map<String, Double> allServerIps = typedTuples.stream()
                    .collect(Collectors.toMap(ZSetOperations.TypedTuple::getValue, ZSetOperations.TypedTuple::getScore));

            double maxMillis = 0;
            String maxMillisServerIp = "";
            for (String serverIp : allServerIps.keySet()) {
                double millis = allServerIps.get(serverIp);
                if (millis >= maxMillis) {
                    maxMillis = millis;
                    maxMillisServerIp = serverIp;
                }
            }

            if (maxMillisServerIp.equals(SERVER_IP)) {
                log.info("[{}] recommend latest launch server ", SERVER_IP);
                refreshRegisterTime();
                return false;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }

}
